# -*- coding: utf-8 -*-
import tornado.ioloop
import tornado.web
from apscheduler.schedulers.background import BackgroundScheduler
from concurrent.futures import ThreadPoolExecutor

from service import *
from chinese_calendar import is_workday
from multiprocessing import  Process
import json
import os
import settings
import sqlite3
import re
import pytz





local_tz = pytz.timezone('Asia/Shanghai')
# 创建一个线程池
executor = ThreadPoolExecutor(max_workers=5)  # 可根据需要调整线程数量

# 初始化 BackgroundScheduler，并设置其 executor 为创建的线程池
scheduler = BackgroundScheduler(executor=executor, timezone=local_tz)

#更新自选
# scheduler.add_job(update_zx_data, trigger="interval", seconds=settings.interval_count)
#更新自选合并
# scheduler.add_job(update_dzj_zx_merge_data, trigger="interval", seconds=settings.interval_count)
#更新大资金
scheduler.add_job(update_dzj_data, trigger="interval", seconds=3)
#更新成交对比
# scheduler.add_job(update_cjdb_data, trigger="interval", seconds=settings.interval_count)
#更新板块
scheduler.add_job(update_bk_data, trigger="interval", seconds=settings.interval_count)
#更新总数据
scheduler.add_job(update_data, trigger="interval", seconds=settings.interval_count)
#更新大盘
scheduler.add_job(update_get_dp, trigger="interval", seconds=settings.interval_count)
#更新板块前10
scheduler.add_job(update_bk10_data, trigger="interval", seconds=settings.interval_count)

#更新内外盘
scheduler.add_job(update_nwp_data, trigger="interval", seconds=3)

# scheduler.add_job(task_update_history, trigger="cron", hour=1, minute=15)
# scheduler.add_job(task_update_volume_ratio, trigger="cron", hour=16)
#更新板块信息
scheduler.add_job(task_update_gn, trigger="cron", hour=10, minute=47)


scheduler.start()


class DpHandler(BaseHandler):
    """
    大盘数据
    """

    @api_auth
    def get(self):
        dp_info = r_con.get("dp_info")
        dp_info = json.loads(dp_info)
        res = {"data": dp_info}
        self.write(response_data(res))


class UpdateZxHandler(BaseHandler):
    """
    更新自选
    """

    @api_auth
    def get(self):
        remark = self.get_argument("remark")
        code = self.get_argument("code")
        sqlite3_conn = sqlite3.connect("stock.db")
        cursor = sqlite3_conn.cursor()
        cursor.execute(f"select * from zxStock where code='{code}'")
        data = cursor.fetchone()
        if not data:
            insert_zx(code)

        sql = f"update zxStock set remark='{remark}' where code='{code}'"
        cursor.execute(sql)
        sqlite3_conn.commit()
        sqlite3_conn.close()
        self.write(response_data({"data": "更新成功"}))


class DeleteZxHandler(BaseHandler):
    """
    删除自选
    """

    @api_auth
    def get(self):
        code = self.get_argument("code")
        sqlite3_conn = sqlite3.connect("stock.db")
        cursor = sqlite3_conn.cursor()
        sql = "delete from zxStock where code='{}'".format(code)
        logger.warning(sql)
        cursor.execute(sql)
        sqlite3_conn.commit()
        sqlite3_conn.close()
        self.write(response_data({"data": "删除成功"}))


class AddZxHandler(BaseHandler):
    """
    添加自选
    """

    @api_auth
    def get(self):
        code = self.get_argument("code")
        insert_zx(code)
        self.write(response_data({"data": "添加成功"}))


class GetZxHandler(BaseHandler):
    """
    获取自选
    """

    @api_auth
    def get(self, zx_type):
        remark = None
        try:
            remark = self.get_argument("value")
        except Exception:
            pass
        settings.current_type = f"zx{zx_type}"
        zx_df_json = r_con.get("zx_data")
        zx_df = pd.read_json(zx_df_json, dtype={"代码": 'str'})
        columns = []
        data = []
        is_kp = check_time()
        if isinstance(zx_df, DataFrame):
            if zx_type == "fp":
                # 复盘
                zx_df = zx_df[zx_df["主力资金"] < 0.3]
            elif zx_type == "tld":
                zx_df = zx_df[zx_df["论股"].str.contains("屠龙")]
            elif zx_type == "zt":
                zx_df = zx_df[zx_df["涨跌幅"] > 9.9]

            if remark:
                zx_df = zx_df[zx_df["论股"].str.contains(remark)]
        update_date = r_con.get("update_date")
        res = {"data": zx_df.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))


class GetDzjHandler(BaseHandler):
    """
    获取大资金
    """

    @api_auth
    def get(self):
        df_json = r_con.get("dzj_data")
        dzj_df = pd.read_json(df_json, dtype={"代码": 'str'})

        settings.current_type = "dzj"
        update_date = r_con.get("update_date")
        res = {"data": dzj_df.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))


class GetCjdbHandler(BaseHandler):
    """
    获取成交对比
    """

    @api_auth
    def get(self):
        df_json = r_con.get("cjdb_data")
        df = pd.read_json(df_json, dtype={"代码": 'str'})

        update_date = r_con.get("update_date")
        res = {"data": df.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))


class GetDzjZxMergeHandler(BaseHandler):
    """
    获取大资金自选合并
    """
    @api_auth
    def get(self):
        zx_type = self.get_argument("type")
        df_json = r_con.get("dzj_zx_merge")
        dzj_df = pd.read_json(df_json, dtype={"代码": 'str'})
        settings.current_type = "dzj_zx_merge"


        if isinstance(dzj_df, DataFrame):
            if zx_type == "rlt":
                dzj_df = dzj_df[dzj_df["涨跌幅"] > 2]

        update_date = r_con.get("update_date")
        res = {"data": dzj_df.to_json(orient='records'),  "update_date": update_date}
        self.write(response_data(res))

class GetZsHandler(BaseHandler):
    """
    获取涨速前20条
    """
    @api_auth
    def get(self):
        df_global = get_filter_data()
        df_global = df_global.head(20)

        update_date = r_con.get("update_date")
        res = {"data": df_global.to_json(orient='records'),"update_date": update_date}
        self.write(response_data(res))

class GetBk10Handler(BaseHandler):
    """
    获取板块前10所有股
    """
    @api_auth
    def get(self):
        bk10_data_json = r_con.get("bk10_data")
        df_bk10 = pd.read_json(bk10_data_json, dtype={"代码": 'str'})
        update_date = r_con.get("update_date")
        res = {"data": df_bk10.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))



class GetWcHandler(BaseHandler):
    """
    获取问财
    """
    @api_auth
    def get(self):
        settings.current_type = "wc"
        df_json = r_con.get("wc1_data")
        wc_df1 = pd.read_json(df_json, dtype={"代码": 'str'})

        df_json = r_con.get("wc2_data")
        wc_df2 = pd.read_json(df_json, dtype={"代码": 'str'})

        df_global = None
        if isinstance(wc_df1, DataFrame) and isinstance(wc_df2, DataFrame):
            wc = pd.concat([wc_df1, wc_df2])
            wc.drop_duplicates(subset=["代码"], keep="first", inplace=True)
            df_global = wc
        elif isinstance(wc_df1, DataFrame):
            df_global = wc_df1
        elif isinstance(wc_df2, DataFrame):
            df_global = wc_df2
            # if is_kp:
            #     wc_df2.sort_values(by="涨速", axis=0, ascending=False, inplace=True)
            # else:
            #     wc_df2.sort_values(by="主力资金", axis=0, ascending=False, inplace=True)

        update_date = r_con.get("update_date")
        res = {"data": df_global.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))


class GetNwpHandler(BaseHandler):
    """
    获取内外盘
    """
    @api_auth
    def get(self):
        df_global_json = r_con.get("nwp_df")
        df_global = pd.read_json(df_global_json, dtype={"代码": 'str'})
        update_date = r_con.get("update_date")
        res = {"data": df_global.to_json(orient='records'), "update_date": update_date}
        self.write(response_data(res))


class GetBkHandler(BaseHandler):
    """
    获取板块信息
    """
    @api_auth
    def get(self):
        # 获取板块信息
        df_json = r_con.get("bk_data")
        if df_json is None:
            df = update_bk_data()
        else:
            df = pd.read_json(df_json, dtype={"代码": 'str'})
        
        columns = df.columns.tolist()
        data = df.values.tolist()
        res = {"data": data, "columns": columns}
        self.write(response_data(res))


class UpdateWcHandler(BaseHandler):
    """
    更新问财数据
    """
    @api_auth
    def post(self, wc_type):
        data = json.loads(self.request.body)
        if data["type"] == "service_err":
            r_con.set("update_date", "问财客户端中断")
        else:
            r_con.setex("heartbeat", 10, 1)
            
        is_kp = check_time()
        def update_data():
            table = None
            if data["type"] != "heartbeat" and data["type"] != "service_err":
                table_new = pd.read_html(data["msg"])
                if table_new:
                    table = table_new[0]

                else:
                    logger.warning("数据为空")
                    r_con.set("update_date", "问财数据为空")
                

            if wc_type == "wc1":
                if isinstance(table, DataFrame):
                    logger.warning("推送策略一..............")
                    table["股票代码"] = table["股票代码"].apply(
                        lambda x: re.findall(r"\b\d{6}\b", x)[0]
                    )
                    wc1_codes = table["股票代码"].tolist()
                    codes_json = json.dumps(wc1_codes)
                    r_con.set("wc1_codes", codes_json)

            elif wc_type == "wc2":
                if isinstance(table, DataFrame):
                    logger.warning("推送策略二..............")
                    table["股票代码"] = table["股票代码"].apply(
                        lambda x: re.findall(r"\b\d{6}\b", x)[0]
                    )
                    wc2_codes = table["股票代码"].tolist()
                    codes_json = json.dumps(wc2_codes)
                    r_con.set("wc2_codes", codes_json)

            elif wc_type == "dzj":
                if isinstance(table, DataFrame):
                    update_date = datetime.strftime(datetime.now(), "%Y-%m-%d %H:%M:%S")
                    r_con.set("update_date", update_date)
                    logger.warning(f"推送主力资金..............{len(table)}")
                    table["股票代码"] = table["股票代码"].apply(
                        lambda x: re.findall(r"\b\d{6}\b", x)[0]
                    )
                    dzj_df_json = table.to_json()
                    r_con.set("dzj_df", dzj_df_json)

            elif wc_type == "cjdb":
                if isinstance(table, DataFrame):
                    logger.warning(f"推送成交对比.............{len(table)}")
                    table["股票代码"] = table["股票代码"].apply(
                        lambda x: re.findall(r"\b\d{6}\b", x)[0]
                    )
                    cjdb_codes = table["股票代码"].tolist()
                    codes_json = json.dumps(cjdb_codes)
                    r_con.set("cjdb_codes", codes_json)
        t1 = Process(target=update_data)
        t1.daemon = True
        t1.start()
        res = {"is_kp": is_kp, "current_type": settings.current_type}
        self.write(response_data(res))

    


def make_app():
    return tornado.web.Application([
        # (r"/member/(\w*)", DpHandler),
        (r"/dpData", DpHandler),
        (r"/updateZxData", UpdateZxHandler),
        (r"/delZxData", DeleteZxHandler),
        (r"/cjdbData", GetCjdbHandler),
        (r"/addZxData", AddZxHandler),
        (r"/zsData", GetZsHandler),
        (r"/zxData/(\w*)", GetZxHandler),
        (r"/dzj", GetDzjHandler),
        (r"/wcData", GetWcHandler),
        (r"/dzjAndZxData", GetDzjZxMergeHandler),
        (r"/bkData", GetBkHandler),
        (r"/bk10Data", GetBk10Handler),
        (r"/nwpData", GetNwpHandler),
        (r"/wc_change/(\w*)", UpdateWcHandler),
    ], static_path=os.path.join(os.path.dirname(__file__), 'static')
    )


app = make_app()
app.listen(address=settings.HOST, port=settings.PORT)
tornado.ioloop.IOLoop.current().start()
